package dk.almbrand.microservice.kafka.consumer;

import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.internal.LinkedTreeMap;

import dk.almbrand.microservice.circuitbreaker.exceptions.ArgumentOutOfRangeException;
import dk.almbrand.microservice.model.ConfirmationEvent;
import dk.almbrand.microservice.model.RequestEvent;
import dk.almbrand.microservice.model.ResponseEvent;
import dk.almbrand.microservice.service.base.MicroserviceWrapper;
import dk.almbrand.microservice.util.GsonUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.ConsumerTimeoutException;
import kafka.consumer.KafkaStream;

import kafka.javaapi.consumer.ConsumerConnector;

import kafka.message.MessageAndMetadata;

import org.I0Itec.zkclient.exception.ZkTimeoutException;

public class ConsumerDaemonImpl implements Runnable, ConsumerDaemon {

    final static String TOPIC = "verification-topic";

    private MicroserviceWrapper wrapper = null;
    private ConsumerConnector consumerConnector = null;
    private String host;
    private String group;
    private String topic;

    public ConsumerDaemonImpl(MicroserviceWrapper wrapper) {
        this.wrapper = wrapper;
    }

    @Override
    public void run() {
        int numOfErrors = 0;

        while (numOfErrors <= 5) {
            try {
                consumerConnector = kafka.consumer.Consumer.createJavaConsumerConnector(getKafkaConfiguration());
                Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
                topicCountMap.put(topic, 1);
                Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumerConnector.createMessageStreams(topicCountMap);
                KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
                ConsumerIterator<byte[], byte[]> it = stream.iterator();
                
                while (it.hasNext()) {
                    ConfirmationEvent confirmationEvent = null;
                    ResponseEvent responseEvent = null;
                    RequestEvent requestEvent = null;
                    
                    MessageAndMetadata<byte[], byte[]> data = it.next();
                    String message = new String(data.message());
                    
                    System.out.println("I base consumer - besked: " + new String(data.message()));
                    
                    JsonObject jsonObj = new Gson().fromJson(message, JsonElement.class).getAsJsonObject();
                    
                    if(jsonObj.get("reserveKey") != null) {
                        confirmationEvent = new GsonUtils<ConfirmationEvent>().fromJSON(message, ConfirmationEvent.class);
                        
                        confirmationEvent.setJson(message);
                        
                        if(confirmationEvent.getObject() instanceof LinkedTreeMap) {
                            confirmationEvent.setObject(GsonUtils.toTreeMap((LinkedTreeMap) confirmationEvent.getObject()));
                        }
                        
                        wrapper.onMessage(confirmationEvent);
                    
                    } else if(jsonObj.get("resultCode") != null) { //if(message.contains("resultCode")) {
                        responseEvent = new GsonUtils<ResponseEvent>().fromJSON(message, ResponseEvent.class);
                        
                        responseEvent.setJson(message);
                        
                        if(responseEvent.getObject() instanceof LinkedTreeMap) {
                            responseEvent.setObject(GsonUtils.toTreeMap((LinkedTreeMap) responseEvent.getObject()));
                        }
                        
                        wrapper.onMessage(responseEvent);
                    } else if(jsonObj.get("responseQueue") != null) { //if(message.contains("responseQueue")) {
                        requestEvent = new GsonUtils<RequestEvent>().fromJSON(message, RequestEvent.class);
                        
                        requestEvent.setJson(message);
                        
                        if(requestEvent.getObject() instanceof LinkedTreeMap) {
                            requestEvent.setObject(GsonUtils.toTreeMap((LinkedTreeMap) requestEvent.getObject()));
                        }
                        
                        wrapper.onMessage(requestEvent);
                    } else {
                        System.out.println("Unknown message: " + new String(data.message()));
                    }
                   
                    // Success - Start over.
                    numOfErrors = 0;
                    System.out.println("10");
                }

            } catch (Exception e) {
                System.out.println(e.getMessage() + " no of errors: " + numOfErrors);
                if (e instanceof ConsumerTimeoutException || e instanceof ZkTimeoutException) {
                    // Hopefully its just a glitch/server busy
                    numOfErrors++;
                    wait(numOfErrors);
                    continue;
                } else {
                    return;
                }
            } finally {
                if (consumerConnector != null) {
                    consumerConnector.shutdown();
                }
            }
        }
        return;
    }

    private ConsumerConfig getKafkaConfiguration() {
        Properties props = new Properties();                
        props.put("zookeeper.connect", host);
        props.put("zookeeper.connection.timeout.ms", "1000000");
        props.put("auto.commit.interval.ms", "1000");
        props.put("group.id", group);
        // props.put("zookeeper.session.timeout.ms", "400");
        // props.put("zookeeper.sync.time.ms", "200");
        // props.put("consumer.timeout.ms", "60000");
        
        return new ConsumerConfig(props);
    }
    
    @SuppressWarnings("oracle.jdeveloper.java.insufficient-catch-block")
    private static void wait(int numOfErrors) {
        try {
            switch (numOfErrors) {
            case 1:
                Thread.sleep(1000); // 1 second
                break;
            case 2:
                Thread.sleep(2000); // 2 seconds
                break;
            case 3:
                Thread.sleep(4000); // 4 seconds
                break;
            case 4:
                Thread.sleep(8000); // 8 seconds
                break;
            case 5:
                Thread.sleep(60000); // 1 minute
                break;
            }
        } catch (InterruptedException e) {
            //Ignore
        }
    }  
    
    public ConsumerDaemon setHost(String host) {
        this.host = host;
        return this;
    }

    public String getHost() {
        return host;
    }

    public ConsumerDaemon setGroup(String group) {
        this.group = group;
        return this;
    }

    public String getGroup() {
        return group;
    }

    public ConsumerDaemon setTopic(String topic) {
        this.topic = topic;
        return this;
    }

    public String getTopic() {
        return topic;
    }
    
    public static void main(String[] args) throws ArgumentOutOfRangeException {
        //new ConsumerImpl().setHost("ubuntu:2181").setGroup("test-group").setTopic(TOPIC).build();
    }
}
